package com.restaurant.service;

import com.restaurant.bean.RestaurantData;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import scala.Tuple2;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.time.LocalDate;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;

@Service
public class DataAnalysisService {
    @Autowired
    private JavaSparkContext sparkContext;

    private JavaRDD<RestaurantData> data;

    @PostConstruct
    public void beforeInit() throws IOException {
        if (!Path.of("./Catering.csv").toFile().exists()) {
            InputStream resourceAsStream = this.getClass().getClassLoader().getResourceAsStream("Catering.csv");
            assert resourceAsStream != null;
            Files.copy(resourceAsStream, Path.of("./Catering.csv"), StandardCopyOption.ATOMIC_MOVE);
            resourceAsStream.close();
        }
        this.data = sparkContext.textFile("./Catering.csv").map((Function<String, RestaurantData>) s -> {
            //类别,行政区,点评数,口味,环境,服务,人均消费,城市,Lng,Lat

            //粤菜,浦东新区,141,6.7,7.2,6.6,141,上海市,121.933142,30.89322409
            String[] split = s.split(",");
            RestaurantData restaurantData = new RestaurantData();
            restaurantData.setType(split[0]);
            restaurantData.setPosition(split[1]);
            restaurantData.setComments(Integer.valueOf(split[2]));
            restaurantData.setFlavor(Double.valueOf(split[3]));
            restaurantData.setEnvironmental(Double.valueOf(split[4]));
            restaurantData.setService(Double.valueOf(split[5]));
            restaurantData.setConsumption(Integer.valueOf(split[6]));
            restaurantData.setCity(split[7]);
            restaurantData.setLng(Double.valueOf(split[8]));
            restaurantData.setLat(Double.valueOf(split[9]));
            restaurantData.setDate(LocalDate.parse(split[10]));
            return restaurantData;
        }).filter((Function<RestaurantData, Boolean>) restaurantData -> !StringUtils.isEmpty(restaurantData.getType())).cache();

        Executors.newFixedThreadPool(1).submit(() -> {
            this.data.collect();
        });
    }

    public Map<String, Long> getTypesWithTotal(String[] date) {
        JavaRDD<RestaurantData> data = this.data;
        if (date.length == 2) {
            data = data.filter((Function<RestaurantData, Boolean>) restaurantData -> {
                LocalDate startDate = LocalDate.parse(date[0]);
                LocalDate endDate = LocalDate.parse(date[1]);
                return restaurantData.getDate().isAfter(startDate) && restaurantData.getDate().isBefore(endDate);
            });
        }
        return data.mapToPair((PairFunction<RestaurantData, String, RestaurantData>) restaurantData -> new Tuple2<>(restaurantData.getType(), restaurantData)).filter((Function<Tuple2<String, RestaurantData>, Boolean>) stringRestaurantDataTuple2 -> !StringUtils.isEmpty(stringRestaurantDataTuple2._1)).countByKey();

    }

    public Double getAllConsumption() {
        return this.data.mapToPair((PairFunction<RestaurantData, Integer, Long>) restaurantData -> new Tuple2<>(1,
            restaurantData.getConsumption().longValue())).groupByKey().mapToPair((PairFunction<Tuple2<Integer, Iterable<Long>>, Integer, Double>) integerIterableTuple2 -> {
            long sum = 0L;
            long count = 0L;
            for (Long aLong : integerIterableTuple2._2()) {
                sum += aLong;
                count++;
            }
            return new Tuple2<>(integerIterableTuple2._1(), 1.0 * sum / count);
        }).collect().get(0)._2();
    }


    public List<Tuple2<Integer, Double>> area() {
        return this.data.mapToPair(restaurantData -> {
            int index = ((int) (restaurantData.getLng() * 10)) % 10;
            return new Tuple2<>(index, restaurantData);
        }).groupByKey().mapToPair(res -> {
            double sum = 0;
            int count = 0;
            for (RestaurantData tmp : res._2()) {
                sum += tmp.getConsumption();
                count++;
            }
            sum = sum / count;
            return new Tuple2<>(res._1(), sum);
        }).collect();
    }


    public List<Tuple2<String, Tuple2>> getLngAndLat(String type) {
        if (StringUtils.isEmpty(type)) {
            return this.data.mapToPair((PairFunction<RestaurantData, String, Tuple2>) restaurantData -> {
                Tuple2<Double, Double> position = new Tuple2<>(restaurantData.getLng(), restaurantData.getLat());
                return new Tuple2<>(restaurantData.getType(), position);

            }).collect();
        } else {
            return this.data.filter((Function<RestaurantData, Boolean>) restaurantData -> restaurantData.getType().equals(type)).mapToPair((PairFunction<RestaurantData, String, Tuple2>) restaurantData -> {
                Tuple2<Double, Double> position = new Tuple2<>(restaurantData.getLng(), restaurantData.getLat());
                return new Tuple2<>(restaurantData.getType(), position);

            }).collect();
        }
    }


    public List<? extends Tuple2> getCostPerformance(String type) {
        if (StringUtils.isEmpty(type)) {
            return this.data.map((Function<RestaurantData, Tuple2<RestaurantData, String>>) restaurantData -> {
                double total = restaurantData.getFlavor() + restaurantData.getEnvironmental() + restaurantData.getService();
                if (total == 0 || restaurantData.getConsumption() == 0) {
                    return new Tuple2<>(restaurantData, "0");
                } else {
                    return new Tuple2<>(restaurantData, String.format("%.2f", total / restaurantData.getConsumption()));
                }
            }).sortBy((Function<Tuple2<RestaurantData, String>, String>) restaurantDataStringTuple2 -> restaurantDataStringTuple2._2, false, 0).take(5000);
        } else if ("all".equals(type)) {
            return this.data.mapToPair((PairFunction<RestaurantData, String, Double>) restaurantData -> {
                double total = restaurantData.getFlavor() + restaurantData.getEnvironmental() + restaurantData.getService();
                if (total == 0 || restaurantData.getConsumption() == 0) {
                    return new Tuple2<>(restaurantData.getType(), 0.0);
                } else {
                    return new Tuple2<>(restaurantData.getType(), total / restaurantData.getConsumption());
                }
            }).groupByKey().filter((Function<Tuple2<String, Iterable<Double>>, Boolean>) stringIterableTuple2 -> !StringUtils.isEmpty(stringIterableTuple2._1())).mapToPair((PairFunction<Tuple2<String, Iterable<Double>>, String, Double>) stringIterableTuple2 -> {
                double sum = 0.0;
                double count = 0;
                for (Double consumption : stringIterableTuple2._2()) {
                    sum += consumption;
                    count++;
                }
                if (count == 0) {
                    return new Tuple2<>(stringIterableTuple2._1(), 0.0);
                } else {
                    return new Tuple2<>(stringIterableTuple2._1(), sum / count);
                }

            }).mapToPair((PairFunction<Tuple2<String, Double>, Double, String>) Tuple2::swap).sortByKey(false).mapToPair((PairFunction<Tuple2<Double, String>
                , String, Double>) Tuple2::swap).collect();

        } else {
            return this.data.filter((Function<RestaurantData, Boolean>) restaurantData -> type.equals(restaurantData.getType())).mapToPair((PairFunction<RestaurantData, String, Double>) restaurantData -> {
                double total = restaurantData.getFlavor() + restaurantData.getEnvironmental() + restaurantData.getService();
                if (restaurantData.getConsumption() == 0) {
                    return new Tuple2<>(restaurantData.getType(), 0.0);
                } else {
                    return new Tuple2<>(restaurantData.getType(), total / restaurantData.getConsumption());
                }

            }).groupByKey().map((Function<Tuple2<String, Iterable<Double>>, Tuple2>) stringIterableTuple2 -> {
                double sum = 0.0;
                double count = 0;
                for (Double consumption : stringIterableTuple2._2()) {
                    sum += consumption;
                    count++;
                }
                if (count == 0) {
                    return new Tuple2<>(stringIterableTuple2._1(), "0");
                } else {
                    return new Tuple2<>(stringIterableTuple2._1(), String.format("%2f", sum / count));
                }
            }).collect();
        }
    }

    public List<String> getTypes() {
        return this.data.map((Function<RestaurantData, String>) restaurantData -> {
            if (!StringUtils.isEmpty(restaurantData.getType())) {
                return restaurantData.getType();
            } else {
                return null;
            }
        }).distinct().filter((Function<String, Boolean>) s -> !StringUtils.isEmpty(s)).collect();
    }
}
